package com.zhilei.deepseekdoctor.utils;

import com.zhilei.deepseekdoctor.enums.SSEMsgType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * SseEmitter 是 Spring 提供的一个类，用来实现 服务端推送（Server-Sent Events，简称 SSE） 的功能。
 * SSE 是一种浏览器与服务器之间的通信机制，允许服务器主动向客户端推送数据，是一种单向的、基于 HTTP 的长连接方案。
 * <p>
 * 基本原理
 * 客户端发起请求后，服务器返回一个 HTTP 响应，Content-Type 设置为 text/event-stream。
 * 服务器不关闭连接，而是持续不断地通过这个连接向客户端发送数据。
 * 客户端使用 JavaScript 的 EventSource 接收这些推送事件。
 *
 * @ClassName SSEServer
 * @Author 王志磊
 * @Version 1.0
 * @Description SSEServer
 **/
@Slf4j
public class SSEServer {

    /**
     * 使用map对象，关联用户id和sse的服务连接,先不管提问，这个提问是后面的事了，先搞一下这个sse
     * 进阶提问1：SseEmitter 能不能放在Redis中和userId进行关联？
     * 进阶提问2：SseEmitter 如何在集群SpringBoot中存在
     */

    // 保存 emitter 实例到内存中（比如 Map 中），以便后续发送数据
    // 线程安全的 HashMap，支持高并发操作，适合在多线程环境中使用。
    // 使用时只需要get需要的key就可以了
    private static final Map<String, SseEmitter> sseClients = new ConcurrentHashMap<>();

    /**
     * 用于统计当前总在线人数
     * 这是定义了一个 线程安全的整数变量，初始值为 0
     * 在多线程环境中（如多个用户同时连接/断开 SSE），会出现线程安全问题，导致统计值不准确，比如多个线程同时执行 onlineCounts++，最终结果可能少加了。
     * 使用 AtomicInteger 能确保加减操作是原子性的，不会发生数据竞争问题。
     * <p>
     * incrementAndGet()	自增 1，并返回新值
     * getAndIncrement()    先返回旧值，在自增 1
     * decrementAndGet()	自减 1，并返回新值
     * get()	获取当前值
     * set(int value)	设置值
     * addAndGet(int x)	增加 x，并返回新值
     */
    private static final AtomicInteger onlineCounts = new AtomicInteger(0);

    /**
     * @param userId
     * @Description: 创建新的SSE连接
     * @Author 王志磊
     */
    public static SseEmitter connect(String userId) {
        // 设置超时时间，0代表永不过期；默认30秒，超时未完成任务则会抛出异常
        SseEmitter sseEmitter = new SseEmitter(0L);

        // SSE连接完成后的回调方法
        // onCompletion() 是 SseEmitter 提供的方法，用于注册一个 连接完成时的回调函数。
        // 当客户端主动关闭连接、网络断开，或者 SseEmitter.complete() 被调用时，Spring 框架会自动触发这个回调函数。
        // onCompletion（Runnable runnable）入参是一个以被执行的任务
        sseEmitter.onCompletion(completionCallback(userId));

        // 发生错误时，打印日志，移除用户
        sseEmitter.onError(errorCallback(userId));

        // 超时回调，打印日志，移除用户
        sseEmitter.onTimeout(timeoutCallback(userId));

        sseClients.put(userId, sseEmitter);
        log.info("当前创建新的SSE连接，用户ID为: {}", userId);

        // 先查询返回值在自增1
        onlineCounts.getAndIncrement();

        // 先自增1再返回值
        // onlineCounts.incrementAndGet();

        return sseEmitter;
    }

    /**
     * @param userId
     * @param message
     * @param msgType
     * @Description: 发送单条消息
     * @Author 王志磊
     */
    public static void sendMessage(String userId, String message, SSEMsgType msgType) {



        // 先查一下这个map是不是空的
        if (CollectionUtils.isEmpty(sseClients)) {
            return;
        }

        // 判断一下这个用户是不是存在，根据key来判断
        if (sseClients.containsKey(userId)) {

            // 从 sseClients 这个连接池中，获取指定 userId 对应的 SSE 长连接对象 SseEmitter
            SseEmitter sseEmitter = sseClients.get(userId);

            // 发送消息
            // 用户的长连接 用户 消息内容 消息类型
            sendEmitterMessage(sseEmitter, userId, message, msgType);
        }
    }

    /**
     * @param message
     * @Description: 发送消息给所有人
     * @Author 王志磊
     */
    public static void sendMessageToAllUsers(String message) {

        // 先查一下这个map是不是空的
        if (CollectionUtils.isEmpty(sseClients)) {
            return;
        }

        // 循环推送
        sseClients.forEach((userId, sseEmitter) -> {
                    sendEmitterMessage(sseEmitter, userId, message, SSEMsgType.MESSAGE);
                }
        );
    }

    /**
     * @param message
     * @Description: 群追加发送消息给所有人
     * @Author 王志磊
     */
    public static void sendMessageAddToAllUsers(String message) {

        // 先查一下这个map是不是空的
        if (CollectionUtils.isEmpty(sseClients)) {
            return;
        }

        // 循环推送
        sseClients.forEach((userId, sseEmitter) -> {
                    sendEmitterMessage(sseEmitter, userId, message, SSEMsgType.ADD);
                }
        );
    }

    /**
     * @param sseEmitter
     * @param userId
     * @param message
     * @param msgType
     * @Description: 使用SseEmitter推送消息
     * @Author 王志磊
     */
    public static void sendEmitterMessage(SseEmitter sseEmitter,
                                          String userId,
                                          String message,
                                          SSEMsgType msgType) {

        try {
            // 用于创建一个 SSE 事件构建器（SseEventBuilder）。
            SseEmitter.SseEventBuilder msg = SseEmitter.event()
                    .id(userId)
                    .name(msgType.type)
                    .data(message);

            // sseEmitter.send(msg) 是将你刚构造好的 SSE 事件通过 HTTP 长连接推送到客户端
            sseEmitter.send(msg);
        } catch (IOException e) {
            log.error("用户[{}]的消息推送发生异常！", userId);
            removeConnection(userId);
        }

    }

    /**
     * @param userId
     * @Description: 主动切断，停止sse服务和客户端的连接
     * @Author 王志磊
     */
    public static void stopServer(String userId) {

        if (CollectionUtils.isEmpty(sseClients)) {
            return;
        }

        SseEmitter sseEmitter = sseClients.get(userId);
        if (sseEmitter != null) {
            // complete 表示执行完毕，断开连接
            sseEmitter.complete();
            removeConnection(userId);
            log.info("连接关闭成功，被关闭的用户为 {}", userId);
        } else {
            log.warn("当前连接无需关闭，请不要重复操作");
        }

    }

    /**
     * @param userId
     * @return Runnable 表示一个可以被执行的任务，没有输入参数，也没有返回值。
     * @Description: SSE连接完成后的回调方法（关闭连接的时候调用）
     * @Author 王志磊
     */
    private static Runnable completionCallback(String userId) {
        return () -> {

            // 记录日志
            log.info("SSE连接完成并结束，用户ID为: {}", userId);

            // 在连接池移除用户信息
            removeConnection(userId);

        };
    }

    /**
     * @param userId
     * @return Runnable
     * @Description: SSE连接超时的时候进行调用
     * @Author 王志磊
     */
    private static Runnable timeoutCallback(String userId) {
        return () -> {
            log.info("SSE连接超时，用户ID为: {}", userId);
            removeConnection(userId);
        };
    }

    /**
     * 这个方法是用来生成一个 SSE连接出错时的回调处理函数，返回的是一个 Consumer<Throwable> 类型的函数接口。
     *
     * @param userId
     * @return Consumer<Throwable> 表示“接受一个参数，没有返回值的函数”。这里它接收的参数类型是 Throwable（即异常对象）。
     * @Description: SSE连接发生错误的时候进行调用
     * @Author 王志磊
     */
    private static Consumer<Throwable> errorCallback(String userId) {
        return Throwable -> {
            // 记录日志
            log.info("SSE连接发生错误，用户ID为: {}", userId);
            // 移除连接。
            removeConnection(userId);
        };
    }

    /**
     * @param userId
     * @Description: 从整个SSE服务中移除用户连接
     * @Author 王志磊
     */
    public static void removeConnection(String userId) {
        sseClients.remove(userId);
        log.info("SSE连接被移除，移除的用户ID为: {}", userId);

        onlineCounts.getAndDecrement();
    }

    /**
     * @param
     * @return int
     * @Description: 获得当前所有的会话总连接数（在线人数）
     * @Author 王志磊
     */
    public static int getOnlineCounts() {
        return onlineCounts.get();
    }

}
